import os
from typing import Tuple
import numpy as np
import pandas as pd
import torch
from torch import nn, optim
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split

# CBAM（Convolutional Block Attention Module） 通道注意力 + 空间注意力
class ChannelAttention(nn.Module):
    def __init__(self, in_planes, ratio=8):
        super(ChannelAttention, self).__init__()
        self.avg_pool = nn.AdaptiveAvgPool2d(1)
        self.max_pool = nn.AdaptiveMaxPool2d(1)
        self.fc = nn.Sequential(
            nn.Conv2d(in_planes, in_planes // ratio, 1, bias=False),
            nn.ReLU(),
            nn.Conv2d(in_planes // ratio, in_planes, 1, bias=False)
        )
        self.sigmoid = nn.Sigmoid()
    def forward(self, x):
        avg_out = self.fc(self.avg_pool(x))
        max_out = self.fc(self.max_pool(x))
        return self.sigmoid(avg_out + max_out)

class SpatialAttention(nn.Module):
    def __init__(self, kernel_size=7):
        super(SpatialAttention, self).__init__()
        self.conv = nn.Conv2d(2, 1, kernel_size, padding=kernel_size//2, bias=False)
        self.sigmoid = nn.Sigmoid()
    def forward(self, x):
        avg_out = torch.mean(x, dim=1, keepdim=True)
        max_out, _ = torch.max(x, dim=1, keepdim=True)
        x_cat = torch.cat([avg_out, max_out], dim=1)
        return self.sigmoid(self.conv(x_cat))

class CBAM(nn.Module):
    def __init__(self, in_planes, ratio=8, kernel_size=7):
        super(CBAM, self).__init__()
        self.channel_attention = ChannelAttention(in_planes, ratio)
        self.spatial_attention = SpatialAttention(kernel_size)
    def forward(self, x):
        out = x * self.channel_attention(x)
        out = out * self.spatial_attention(out)
        return out

# CNN + Attention模型定义
class SimpleCNN_CBAM(nn.Module):
    def __init__(self):
        super(SimpleCNN_CBAM, self).__init__()
        self.conv1 = nn.Sequential(
            nn.Conv2d(1, 16, kernel_size=3, padding=1),
            nn.ReLU(),
            CBAM(16),
            nn.MaxPool2d(2)
        )
        self.conv2 = nn.Sequential(
            nn.Conv2d(16, 32, kernel_size=3, padding=1),
            nn.ReLU(),
            CBAM(32),
            nn.MaxPool2d(2)
        )
        self.conv3 = nn.Sequential(
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.ReLU(),
            CBAM(64),
            nn.MaxPool2d(2)
        )
        self.fc_layers = nn.Sequential(
            nn.Linear(64 * 16 * 16, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 2)
        )

    def forward(self, x):
        x = self.conv1(x)  # 128→64
        x = self.conv2(x)  # 64→32
        x = self.conv3(x)  # 32→16
        x = x.view(x.size(0), -1)
        x = self.fc_layers(x)
        return x


def get_train_data(label_csv: str, svs_feature_path: str) -> Tuple[np.ndarray, np.ndarray]:
    label_df = pd.read_csv(label_csv, encoding='utf-8')  # slide_id,label
    svs_feature = []
    svs_label = []
    for _, row in label_df.iterrows():
        svs_name = row["slide_id"]
        label = row["label"]
        feat_path = os.path.join(svs_feature_path, f"{svs_name}.csv")
        feat = np.loadtxt(feat_path, delimiter=",")  # shape: [128, 128]
        svs_feature.append(feat)
        svs_label.append(label)
    X = np.array(svs_feature)
    Y = np.array(svs_label)
    return X, Y


def train_svs_classifier(feature_dir: str, label_csv: str,
                         epochs: int = 30, batch_size: int = 8,
                         lr: float = 1e-4, model_save_path: str = "svs_transfer_model_cnn_att.pth"):
    # 数据准备
    X, Y = get_train_data(label_csv, feature_dir)
    X_train, X_val, y_train, y_val = train_test_split(X, Y, test_size=0.3, stratify=Y, random_state=42)
    print(f"训练集正负样本分布: {np.bincount(y_train)}")
    print(f"验证集正负样本分布: {np.bincount(y_val)}")

    # 转为 Tensor，CNN输入需要 4D：[batch_size, 1, 128, 128]
    X_train_tensor = torch.tensor(X_train, dtype=torch.float32).unsqueeze(1)
    X_val_tensor = torch.tensor(X_val, dtype=torch.float32).unsqueeze(1)
    y_train_tensor = torch.tensor(y_train, dtype=torch.long)
    y_val_tensor = torch.tensor(y_val, dtype=torch.long)

    train_loader = DataLoader(TensorDataset(X_train_tensor, y_train_tensor), batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(TensorDataset(X_val_tensor, y_val_tensor), batch_size=batch_size)

    # 模型定义
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"Using device: {device}")
    model = SimpleCNN_CBAM().to(device)
    optimizer = optim.Adam(model.parameters(), lr=lr)
    criterion = nn.CrossEntropyLoss()

    # 开始训练
    for epoch in range(epochs):
        model.train()
        total_loss = 0
        for xb, yb in train_loader:
            xb, yb = xb.to(device), yb.to(device)
            optimizer.zero_grad()
            preds = model(xb)
            loss = criterion(preds, yb)
            loss.backward()
            optimizer.step()
            total_loss += loss.item()

        # 验证
        model.eval()
        correct = 0
        total = 0
        with torch.no_grad():
            for xb, yb in val_loader:
                xb, yb = xb.to(device), yb.to(device)
                preds = model(xb)
                pred_labels = preds.argmax(dim=1)
                correct += (pred_labels == yb).sum().item()
                total += yb.size(0)

        print(f"Epoch {epoch + 1}: Loss = {total_loss:.4f}, Val Acc = {correct / total:.4f}")

    # 保存模型
    torch.save(model.state_dict(), model_save_path)
    print(f"✅ 模型已保存到 {model_save_path}")


if __name__ == '__main__':
    train_svs_classifier(
        feature_dir="D:/zm_scientific_research/svs_features",
        label_csv="D:/zm_scientific_research/transfer/label.csv",
        model_save_path="D:/zm_scientific_research/transfer/svs_transfer_model_cnn_att.pth",
        epochs=300,
        batch_size=2
    )
